package com.cssw.fyzb.busi.service.live;

import com.alibaba.fastjson.JSON;
import com.cssw.fyzb.busi.cache.Cache;
import com.cssw.fyzb.busi.cache.CacheKeys;
import com.cssw.fyzb.busi.config.EventClient;
import com.cssw.fyzb.busi.config.ParamConfig;
import com.cssw.fyzb.busi.config.TpsClient;
import com.cssw.fyzb.busi.convert.Converter;
import com.cssw.fyzb.busi.model.tps.Address;
import com.cssw.fyzb.busi.model.tps.LiveStream;
import com.cssw.fyzb.busi.model.tps.OptLiveStream;
import com.cssw.fyzb.core.model.vo.live.*;
import com.cssw.fyzb.busi.util.BizUtil;
import com.cssw.fyzb.busi.util.DateUtil;
import com.cssw.fyzb.busi.util.TimeUtil;
import com.cssw.fyzb.core.dao.live.AnchorDao;
import com.cssw.fyzb.core.dao.live.LiveDao;
import com.cssw.fyzb.core.dao.live.LivePkDao;
import com.cssw.fyzb.core.dao.live.LiveReportDao;
import com.cssw.fyzb.core.mao.live.LivePkMao;
import com.cssw.fyzb.core.mao.live.LiveStatMao;
import com.cssw.fyzb.core.mao.live.UserLiveMao;
import com.cssw.fyzb.core.mao.user.UserFollowMao;
import com.cssw.fyzb.core.mao.user.UserMao;
import com.cssw.fyzb.core.model.dodb.*;
import com.cssw.fyzb.core.model.mos.*;
import com.cssw.fyzb.core.model.vo.live.RankItemVo;
import com.cssw.fyzb.core.model.vo.live.RankVo;
import com.cssw.fyzb.core.util.StringUtil;
import com.ydn.dbframe.plugin.activerecord.Page;
import com.ydn.simplecache.SimpleCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

import static com.cssw.fyzb.busi.cache.CacheKeys.anchorBlackKey;
import static com.cssw.fyzb.busi.cache.CacheKeys.liveKickOutKey;
import static com.cssw.fyzb.busi.util.Const.LIVE_IDLE;
import static com.cssw.fyzb.busi.util.Const.LIVE_LIVING;

/**
 * @author Feng Chen
 */
@Service
public class LiveServiceImpl implements LiveService {

    @Autowired
    private LiveDao liveDao;

    @Autowired
    private LivePkDao livePkDao;

    @Autowired
    private LiveReportDao liveReportDao;

    @Autowired
    private AnchorDao anchorDao;

    @Autowired
    private TpsClient tpsClient;

    @Autowired
    private UserLiveMao userLiveMao;

    @Autowired
    private LivePkMao livePkMao;

    @Autowired
    private UserMao userMao;

    @Autowired
    private UserFollowMao userFollowMao;

    @Autowired
    private SimpleCache simpleCache;

    @Autowired
    private LiveStatMao liveStatMao;

    @Autowired
    private EventClient eventClient;

    @Autowired
    private ParamConfig paramConfig;


    private static final String APP_FYZB = "live";
    private static final Integer PK_TIME = 5;

    private static final Logger logger = LoggerFactory.getLogger(LiveServiceImpl.class);

    @Override
    public CmfLive findByLiveId(Long liveId) {
        return liveDao.findOne(liveId);
    }

    @Override
    public CmfLive getLivingByUserId(Long userId) {
        return liveDao.getLivingByUserId(userId);
    }

    @Override
    public LiveRoomVo createRoom(Long userId, Long anchorId, String title, String thumb, String lat, String lng) {

        // 获取地址
        Address address = tpsClient.getAddress(lat, lng);
        if (null == address){
            address = new Address();
        }

        // 创建直播流
        LiveStream stream = tpsClient.createLiveStream(APP_FYZB, String.valueOf(anchorId), null);

        // 创建房间信息
        Long liveId = liveDao.createRoom(
                userId, anchorId, title, thumb, stream.getPushUrl(), stream.getPullUrl(), stream.getAccelerateUrl(),
                lat, lng, address.getProvince(), address.getCity()
        );

        liveStatMao.create(liveId, userId, anchorId, stream.getPullUrl(), stream.getPushUrl(), stream.getAccelerateUrl());

        LiveRoomVo vo = new LiveRoomVo();
        vo.setLiveId(liveId);
        vo.setPushUrl(stream.getPushUrl());
        vo.setPullUrl(stream.getPullUrl());
        vo.setAccelerateUrl(stream.getAccelerateUrl());
        vo.setAnchorId(anchorId);

        // 修改直播状态
        anchorDao.updateLiveByAnchor(anchorId, LIVE_LIVING);

        //创建直播间
        userLiveMao.create(liveId);

        //更新用户正在直播中
        userMao.updateLive(userId, "1");

        return vo;
    }

    @Override
    public void dropRoom(Long anchorId, Long liveId) {
        // 关闭直播流
        OptLiveStream opt = tpsClient.dropLiveStream(APP_FYZB, String.valueOf(anchorId));
        //直播流如果已经关闭了，把本地直播也关闭
        if (opt.getErrCode() != null && !"ResourceNotFound.StreamNotAlive".equals(opt.getErrCode())) {
            logger.error("DropLiveStream errCode: {}, errMsg: {}", opt.getErrCode(), opt.getErrMsg());
        } else {

            // 更新主播状态
            anchorDao.updateLiveByAnchor(anchorId, LIVE_IDLE);

            // 删除房间信息
            liveDao.dropRoom(liveId);

            LiveStatMo liveStatMo = liveStatMao.findByLiveId(liveId);
            //更新直播状态
            userMao.updateLive(liveStatMo.getUserId(), "0");
            liveStatMao.drop(liveId);

        }
    }

    @Override
    public EnterLiveRoomVo enterRoom(Long liveId, Long userId) {
        LiveStatMo liveStatMo = liveStatMao.findByLiveId(liveId);
        boolean isBlack = isAnchorBlack(liveStatMo.getAnchorId(), userId);
        if (isBlack){
            BizUtil.exception("你已被拉黑，进入房间失败");
        }

        boolean isKickOut = isLiveKickOut(liveId, userId);
        if (isKickOut){
            BizUtil.exception("你已被踢出进入房间失败");
        }

        boolean exists = userLiveMao.exists(liveId, userId);
        if (!exists) {
            liveStatMao.incViews(liveId);
            userLiveMao.enter(liveId, userId);
        } else {
            userLiveMao.reEnter(liveId, userId);
        }

        EnterLiveRoomVo vo = new EnterLiveRoomVo();
        vo.setLiveId(liveId);
        vo.setUserId(liveStatMo.getUserId());
        vo.setFollow(userFollowMao.exists(userId, liveStatMo.getUserId()) ? "1": "0");

        UserMo userMo = userMao.findByUserId(liveStatMo.getUserId());
        vo.setAvatar(userMo.getAvatar());
        vo.setNickName(userMo.getNickName());
        vo.setLikes(liveStatMo.getLikes());

        //红包是有效的，才返回
        if (null != liveStatMo.getRedOpenTime()
                && liveStatMo.getRedOpenTime() > System.currentTimeMillis()) {
            vo.setRedPackId(liveStatMo.getRedPackId());
            vo.setRedOpenTime(liveStatMo.getRedOpenTime());
        }

        vo.setPk(liveStatMo.getPk());
        if ("1".equals(liveStatMo.getPk())){
            UserMo pkUserMo = userMao.findByUserId(liveStatMo.getPkUserId());
            vo.setPkUserId(pkUserMo.getUserId());
            vo.setPkAvatar(pkUserMo.getAvatar());
            vo.setPkNickName(pkUserMo.getNickName());
            vo.setPkLiveId(liveStatMo.getPkLiveId());

            List<LivePkMo> livePkMos = livePkMao.findByPk(liveStatMo.getPkId());
            if (null != livePkMos){
                for (LivePkMo livePkMo: livePkMos){
                    RankVo cache = Cache.get(CacheKeys.pkRankKey(livePkMo.getLiveId()));
                    if (livePkMo.getLiveId().toString().equals(liveId.toString())){
                        vo.setPkFromTotalGift(livePkMo.getTotal());
                        vo.setPkFromAudience(convertTopAudiences(cache.getList()));
                    } else {
                        vo.setPkToTotalGift(livePkMo.getTotal());
                        vo.setPkToAudience(convertTopAudiences(cache.getList()));
                    }
                }
            }
        }

        //获取当前主播的观众信息
        RankVo cache = Cache.get(CacheKeys.rewardRankKey(liveId));
        if (null != cache) {
            vo.setTotalAudience(cache.getTotal());
            vo.setTopAudiences(convertTopAudiences(cache.getList()));
        } else {
            UserLiveMo userLiveMo = userLiveMao.findByLiveId(liveId);
            List<AudienceVo> list = ((Converter<List<AudienceMo>, List<AudienceVo>>) audiences -> {
                List<AudienceVo> vos = new ArrayList<>();
                if (null == audiences || audiences.isEmpty()) {
                    return vos;
                }
                audiences.sort(Comparator.comparing(AudienceMo::getReward, Comparator.nullsLast(Integer::compareTo)).reversed());
                if (audiences.size() > 3){
                    return convertAudienceVo(audiences.subList(0, 3));
                }
                return convertAudienceVo(audiences);
            }).convert(userLiveMo.getAudiences());

            vo.setTopAudiences(list);
            vo.setTotalAudience(0);
            if (null != userLiveMo.getAudiences()){
                vo.setTotalAudience(userLiveMo.getAudiences().size());
            }
        }

        reRank(liveId);

        return vo;
    }

    private boolean isLiveKickOut(Long liveId, Long userId) {
        List<CmfLiveKickOut> list = getLiveKickOutByLiveId(liveId);

        if (null == list){
            return false;
        }
        for (CmfLiveKickOut kickOut: list){
            if (kickOut.getUserId().toString().equals(userId.toString())){
                return true;
            }
        }

        return false;
    }


    private List<CmfLiveKickOut> getLiveKickOutByLiveId(Long liveId){
        String cacheStr = Cache.get(liveKickOutKey(liveId));

        List<CmfLiveKickOut> kickOuts = null;
        if (StringUtil.isEmpty(cacheStr)){
            kickOuts = liveDao.listLiveKickOut(liveId);

            if (null == kickOuts){
                kickOuts = new ArrayList<>();
            }
            Cache.put(liveKickOutKey(liveId), JSON.toJSONString(kickOuts));
        } else{
            kickOuts = JSON.parseArray(cacheStr, CmfLiveKickOut.class);
        }

        return kickOuts;
    }

    private boolean isAnchorBlack(Long anchorId, Long userId) {
        List<CmfAnchorBlacklist> list = getAnchorBlackByAnchorId(anchorId);

        if (null == list){
            return false;
        }
        for (CmfAnchorBlacklist blacklist: list){
            if (blacklist.getUserId().toString().equals(userId.toString())){
                return true;
            }
        }

        return false;
    }

    private List<CmfAnchorBlacklist> getAnchorBlackByAnchorId(Long anchorId){
        String cacheStr = Cache.get(anchorBlackKey(anchorId));

        List<CmfAnchorBlacklist> blacklists = null;
        if (StringUtil.isEmpty(cacheStr)){
            blacklists = anchorDao.listAnchorBlacklist(anchorId);

            if (null == blacklists){
                blacklists = new ArrayList<>();
            }
            Cache.put(anchorBlackKey(anchorId), JSON.toJSONString(blacklists));
        } else{
            blacklists = JSON.parseArray(cacheStr, CmfAnchorBlacklist.class);
        }

        return blacklists;
    }
    private void reRank(Long liveId){
        Map<String, Object> param = new HashMap<>();
        param.put("liveId", liveId);
        eventClient.fireEvent("reRank", JSON.toJSONString(param));
    }

    private List<AudienceVo> convertTopAudiences(List<RankItemVo> rankItemVos){
        List<AudienceVo> vos = new ArrayList<>();
        if (rankItemVos == null || rankItemVos.isEmpty()){
            return vos;
        }
        AudienceVo vo = null;
        for (RankItemVo itemVo: rankItemVos){
            vo = new AudienceVo();
            vo.setUserId(itemVo.getUserId());
            vo.setRank(itemVo.getNo());

            UserMo userMo = userMao.findByUserId(itemVo.getUserId());
            if (null != userMo) {
                vo.setAvatar(userMo.getAvatar());
                vo.setNickName(userMo.getNickName());
            }
            vos.add(vo);
        }
        return vos;
    }

    @Override
    public void exitRoom(Long liveId, Long userId) {
        boolean exists = userLiveMao.exists(liveId, userId);
        if (exists) {
            userLiveMao.exit(liveId, userId);
        }

        Map<String, Object> param = new HashMap<>();
        param.put("liveId", liveId);
        eventClient.fireEvent("reRank", JSON.toJSONString(param));

    }

    @Override
    public List<AudienceVo> listAudience(Long liveId, Long userId, Integer num) {
        UserLiveMo userLiveMo = userLiveMao.findByLiveId(liveId);

        List<AudienceVo> list = ((Converter<List<AudienceMo>, List<AudienceVo>>) audiences -> {
            List<AudienceVo> vos = new ArrayList<>();
            if (null == audiences || audiences.isEmpty()) {
                return vos;
            }
            audiences.sort(Comparator.comparing(AudienceMo::getReward, Comparator.nullsLast(Integer::compareTo)).reversed());

            int rank = 1;
            int selfIndex = 0;
            for (AudienceMo v : audiences) {
                AudienceVo vo = new AudienceVo();
                UserMo userMo = userMao.findByUserId(v.getUserId());
                if (null != userMo) {
                    vo.setNickName(userMo.getNickName());
                    vo.setAvatar(userMo.getAvatar());
                }
                if (StringUtil.isEmpty(userMo.getAvatar())){
                    vo.setAvatar(paramConfig.getAvatar());
                }

                vo.setRank(rank);
                vo.setUserId(v.getUserId());
                vo.setReward(v.getReward());
                if (userId.longValue() == v.getUserId().longValue()) {
                    selfIndex = rank - 1;
                }
                rank++;
                vos.add(vo);
            }

            //如果传入了数量
            if (null != num) {
                if (audiences.size() < num) {
                    return vos;
                } else {
                    List<AudienceVo> returnVos = vos.subList(0, num);
                    if (num <= selfIndex) {
                        returnVos.add(vos.get(selfIndex));
                    }
                    return returnVos;
                }
            }
            return vos;
        }).convert(userLiveMo.getAudiences());

        return list;
    }


    @Override
    public List<AudienceVo> listAudienceByAnchor(Long liveId) {
        UserLiveMo userLiveMo = userLiveMao.findByLiveId(liveId);
        List<AudienceVo> list = ((Converter<List<AudienceMo>, List<AudienceVo>>) audiences -> {
            List<AudienceVo> vos = new ArrayList<>();
            if (null == audiences || audiences.isEmpty()) {
                return vos;
            }
            audiences.sort(Comparator.comparing(AudienceMo::getReward, Comparator.nullsLast(Integer::compareTo)).reversed());

            return convertAudienceVo(audiences);
        }).convert(userLiveMo.getAudiences());

        return list;
    }

    private List<AudienceVo> convertAudienceVo(List<AudienceMo> audiences) {
        List<AudienceVo> vos = new ArrayList<>();
        int rank = 1;
        for (AudienceMo v : audiences) {
            AudienceVo vo = new AudienceVo();
            UserMo userMo = userMao.findByUserId(v.getUserId());
            if (null != userMo) {
                vo.setNickName(userMo.getNickName());
                vo.setAvatar(userMo.getAvatar());
            }
            if (StringUtil.isEmpty(userMo.getAvatar())){
                vo.setAvatar(paramConfig.getAvatar());
            }

            vo.setRank(rank);
            vo.setUserId(v.getUserId());
            vo.setReward(v.getReward());
            rank++;
            vos.add(vo);
        }

        return vos;
    }

    @Override
    public CmfLiveKickOut queryLiveKickOut(Long liveId, Long operUserId, Long userId) {
        return liveDao.queryLiveKickOut(liveId, operUserId, userId);
    }

    @Override
    public Long addLiveKickOut(Long liveId, Long operUserId, Long userId) {
        Long koId = liveDao.addLiveKickOut(liveId, operUserId, userId);
        simpleCache.delete(liveKickOutKey(liveId));
        return koId;
    }

    @Override
    public void removeLiveKickOutById(Long id) {
        CmfLiveKickOut kickOut = liveDao.queryLiveKickOutById(id);
        if (null != kickOut) {
            liveDao.removeLiveKickOutById(id);
            simpleCache.delete(liveKickOutKey(kickOut.getLiveId()));
        }
    }

    @Override
    public Page<CmfLiveKickOut> listLiveKickOut(Long liveId, Integer pageNumber, Integer pageSize) {
        return liveDao.listLiveKickOut(liveId, pageNumber, pageSize);
    }

    @Override
    public CmfLiveMute queryLiveMute(Long liveId, Long operUserId, Long userId) {
        return liveDao.queryLiveMute(liveId, operUserId, userId);
    }

    @Override
    public Long addLiveMute(Long liveId, Long operUserId, Long userId) {
        LiveStatMo liveStat = liveStatMao.findByLiveId(liveId);
        if (null == liveStat) {
            BizUtil.exception("直播不存在");
        }
        if ("1".equals(liveStat.getStatus())) {
            BizUtil.exception("直播已关闭");
        }
        if (operUserId.toString().equals(liveStat.getUserId().toString())) {
            BizUtil.exception("您不是主播，禁言失败");
        }

        return liveDao.addLiveMute(liveId, operUserId, userId);
    }

    @Override
    public void removeLiveMuteById(Long id) {
        liveDao.removeLiveMuteById(id);
    }

    @Override
    public Page<CmfLiveMute> listLiveMute(Long liveId, Integer pageNumber, Integer pageSize) {
        return liveDao.listLiveMute(liveId, pageNumber, pageSize);
    }

    @Override
    public Long addLiveReport(Long liveId, Long fromUserId, Long toUserId, String content) {
        return liveReportDao.addLiveReport(liveId, fromUserId, toUserId, content);
    }

    @Override
    public void removeLiveReportById(Long id) {
        liveReportDao.removeLiveReportById(id);
    }

    @Override
    public Page<CmfLiveReport> listLiveReport(Long liveId, Integer pageNumber, Integer pageSize) {
        return liveReportDao.listLiveReport(liveId, pageNumber, pageSize);
    }

    @Override
    public Long addLivePk(Long userId, Long fromLiveId, Long toLiveId) {
        if (fromLiveId.toString().equals(toLiveId.toString())){
            BizUtil.exception("不能和自己pk");
        }

        Date now = new Date();

        // pk 五分钟
        Date expire = DateUtil.addMinute(now, PK_TIME);

        LiveStatMo fromLiveStatMo = liveStatMao.findByLiveId(fromLiveId);
        LiveStatMo toLiveStatMo = liveStatMao.findByLiveId(toLiveId);
        if (null == fromLiveStatMo || null == toLiveStatMo){
            BizUtil.exception("直播间不存在");
        }

        liveDao.updateLianmai(fromLiveId, CmfLive.LIANMAI_YES);
        liveDao.updateLianmai(toLiveId, CmfLive.LIANMAI_YES);

        Long pkId = livePkDao.addLivePk(fromLiveStatMo.getAnchorId(), fromLiveId,
                toLiveStatMo.getAnchorId(), toLiveId, now);

        liveStatMao.startPk(fromLiveId, pkId, toLiveId, toLiveStatMo.getUserId());
        liveStatMao.startPk(toLiveId, pkId, fromLiveId, fromLiveStatMo.getUserId());

        // 将 pk 详情放到缓存中
        LivePkVo from = new LivePkVo();
        from.setPkId(pkId);
        from.setLiveId(fromLiveId);
        from.setExpire(expire.getTime());
        Cache.put(CacheKeys.pkingKey(fromLiveId), from);
        // 添加 mongo
        livePkMao.create(fromLiveId, pkId);

        LivePkVo to = new LivePkVo();
        to.setPkId(pkId);
        to.setLiveId(fromLiveId);
        to.setExpire(expire.getTime());
        Cache.put(CacheKeys.pkingKey(toLiveId), to);
        livePkMao.create(toLiveId, pkId);

        return pkId;
    }

    @Override
    public void closeLivePk(Long pkId, Long userId) {
        CmfLivePk livePk = livePkDao.findById(pkId);

        livePkDao.closeLivePk(pkId);

        //结束pk
        liveStatMao.stopPk(livePk.getFromLiveId());
        liveStatMao.stopPk(livePk.getToLiveId());

        liveDao.updateLianmai(livePk.getFromLiveId(), CmfLive.LIANMAI_NO);
        liveDao.updateLianmai(livePk.getToLiveId(), CmfLive.LIANMAI_NO);

        // 删除缓存
        Cache.remove(CacheKeys.pkingKey(livePk.getFromLiveId()));
        Cache.remove(CacheKeys.pkingKey(livePk.getToLiveId()));

        //todo 保存pk数据

    }

    @Override
    public Page<LiveFriendVo> listLiveFriend(Long userId, Integer pageNumber, Integer pageSize) {
        Page<CmfLive> page = liveDao.listLiveFriend(userId, pageNumber, pageSize);

        List<LiveFriendVo> list = ((Converter<List<CmfLive>, List<LiveFriendVo>>) lives -> {
            List<LiveFriendVo> vos = new ArrayList<>();

            if (lives != null) {
                for (CmfLive live : lives) {
                    LiveFriendVo vo = new LiveFriendVo();
                    UserMo userMo = userMao.findByUserId(live.getUserId());
                    vo.setUserId(userMo.getUserId());
                    vo.setNickName(userMo.getNickName());
                    vo.setAvatar(userMo.getAvatar());
                    vo.setSex(userMo.getSex());
                    vo.setLiveId(live.getLiveId());
                    UserLiveMo userLiveMo = userLiveMao.findByLiveId(live.getLiveId());
                    if (null == userLiveMo
                            || userLiveMo.getAudiences() == null
                            || userLiveMo.getAudiences().isEmpty()) {
                        vo.setViewers(0);
                    } else {
                        vo.setViewers(userLiveMo.getAudiences().size());
                    }

                    vos.add(vo);
                }
            }
            return vos;
        }).convert(page.getList());

        return new Page<>(list, pageNumber, pageSize, page.getTotalPage(), page.getTotalRow());
    }

    @Override
    public LiveMatchPkVo sameCityMatchPk(Long userId) {
        CmfLive live = liveDao.getLivingByUserId(userId);
        if (StringUtil.isEmpty(live.getCity())) {
            BizUtil.exception("未匹配到主播");
        }

        Map<String, Object> param = new HashMap<>();
        param.put("city", live.getCity());
        param.put("status", CmfLive.STATUS_LIVING);
        param.put("userId", live.getUserId());

        CmfLive matchLive = matchLive(param);
        if (null == matchLive) {
            BizUtil.exception("未匹配到主播");
        }

        LiveMatchPkVo vo = new LiveMatchPkVo();
        vo.setLiveId(matchLive.getLiveId());
        vo.setUserId(matchLive.getUserId());
        UserMo userMo = userMao.findByUserId(live.getUserId());
        vo.setNickName(userMo.getNickName());
        vo.setAvatar(userMo.getAvatar());
        vo.setSex(userMo.getSex());
        return vo;
    }

    @Override
    public LiveMatchPkVo nationwideMatchPk(Long userId) {
        CmfLive live = liveDao.getLivingByUserId(userId);
        if (StringUtil.isEmpty(live.getCity())) {
            BizUtil.exception("未匹配到主播");
        }

        Map<String, Object> param = new HashMap<>();
        param.put("status", CmfLive.STATUS_LIVING);
        param.put("userId", live.getUserId());

        CmfLive matchLive = matchLive(param);
        if (null == matchLive) {
            BizUtil.exception("未匹配到主播");
        }

        LiveMatchPkVo vo = new LiveMatchPkVo();
        vo.setLiveId(matchLive.getLiveId());
        vo.setUserId(matchLive.getUserId());
        UserMo userMo = userMao.findByUserId(live.getUserId());
        vo.setNickName(userMo.getNickName());
        vo.setAvatar(userMo.getAvatar());
        vo.setSex(userMo.getSex());

        return vo;
    }

    @Override
    public Page<LiveVo> searchLive(String key, Integer pageNumber, Integer pageSize) {
        Map<String, Object> param = new HashMap<>();
        param.put("key", key);
        param.put("status", CmfLive.STATUS_LIVING);

        Page<CmfLive> page = liveDao.searchLive(param, pageNumber, pageSize);

        List<LiveVo> list = ((Converter<List<CmfLive>, List<LiveVo>>) lives -> {
            List<LiveVo> vos = new ArrayList<>();

            lives.forEach((v) -> {
                LiveVo vo = new LiveVo();
                vo.setLiveId(v.getLiveId());
                vo.setAnchorId(v.getAnchorId());
                vo.setUserId(v.getUserId());
                vo.setThumb(v.getThumb());
                vo.setPull(v.getPull());

                vo.setLiveTimeStr(TimeUtil.getIntervalStr(v.getStartTime()));

                UserMo userMo = userMao.findByUserId(v.getUserId());
                vo.setNickName(userMo.getNickName());
                vo.setAvatar(userMo.getAvatar());

                vos.add(vo);
            });
            return vos;
        }).convert(page.getList());

        return new Page<>(list, pageNumber, pageSize, page.getTotalPage(), page.getTotalRow());

    }

    @Override
    public void liveLike(String liveId, String userId) {
        Map<String, Object> param = new HashMap<>();
        param.put("liveId", liveId);
        param.put("userId", userId);
        eventClient.fireEvent("liveLike", JSON.toJSONString(param));
    }

    @Override
    public LiveStatVo getLiveStat(Long liveId, Long userId) {
        LiveStatMo liveStatMo = liveStatMao.findByLiveId(liveId);
        LiveStatVo vo = new LiveStatVo();
        if (null == liveStatMo){
            return null;
        }
        vo.setLiveId(liveId);
        vo.setUserId(liveStatMo.getUserId());
        vo.setLikes(liveStatMo.getLikes());
        vo.setViews(liveStatMo.getViews());
        UserMo userMo = userMao.findByUserId(liveStatMo.getUserId());
        if (null != userMo){
            vo.setAvatar(userMo.getAvatar());
            vo.setNickName(userMo.getNickName());
        }
        vo.setFollow(userFollowMao.exists(userId, liveStatMo.getUserId()) ? "1": "0");
        long liveTimeMills = 0L;
        if (null != liveStatMo.getCompleteTime()){
            liveTimeMills = liveStatMo.getCompleteTime().getTime() - liveStatMo.getLiveTime().getTime();
        } else {
            liveTimeMills = System.currentTimeMillis() - liveStatMo.getLiveTime().getTime();
        }

        long nh = 1000 * 60 * 60;// 一小时的毫秒数
        long nm = 1000 * 60;// 一分钟的毫秒数
        long ns = 1000;// 一秒钟的毫秒数
        long hour = liveTimeMills / nh;
        long minute = liveTimeMills % nh / nm;
        long second = liveTimeMills % nh % nm / ns;
        String liveTimeStr = hour + ":" + minute + ":" + second;
        vo.setLiveTimeSt(liveTimeStr);
        return vo;
    }

    @Override
    public UserLiveMo getUserLive(Long liveId) {
        return userLiveMao.findByLiveId(liveId);
    }

    private CmfLive matchLive(Map param) {
        Integer pageNumber = 1;
        Integer pageSize = 20;
        int count = 0;
        while (true) {
            Page<CmfLive> page = liveDao.randomPkLive(param, pageNumber, pageSize);
            if (page.getList() == null || page.getList().isEmpty()) {
                break;
            }

            //最多随机匹配四次
            count++;
            if (count > 4){
                return null;
            }

            for (CmfLive cmfLive : page.getList()) {
                //已经在连麦跳过
                if (cmfLive.isLianmai()) {
                    continue;
                }

                //在连麦中也跳过
                String cacheLianmai = simpleCache.get(CacheKeys.forLianmai(cmfLive.getLiveId()));
                if (StringUtil.isNotEmpty(cacheLianmai)) {
                    continue;
                }

                //只缓存30秒
                simpleCache.put(CacheKeys.forLianmai(cmfLive.getLiveId()), "1", 30);

                return cmfLive;
            }
        }
        return null;
    }
}

